package dm.ass2.user;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

import dm.ass2.user.io.CBElement;
import dm.ass2.user.io.UserMapOutputValueWritable;

public class UserCombiner extends Reducer<IntWritable, UserMapOutputValueWritable, IntWritable, UserMapOutputValueWritable> {

	@Override
	protected void reduce(IntWritable key, Iterable<UserMapOutputValueWritable> value, Context context) throws IOException, InterruptedException {
		List<Integer> usersList = new LinkedList<Integer>();
		Map<Integer,CBElement> cbElements = new HashMap<Integer, CBElement>();
		
		for (UserMapOutputValueWritable user : value) {
			for (Integer userId : user.getUsersList()) {
				usersList.add(userId);
			}
			
			for (int clusterId = 0; clusterId < user.getCbElements().length; clusterId++) {
				CBElement cbElement = cbElements.get(clusterId);
				if (cbElement == null) {
					cbElement = user.getCbElements()[clusterId];
					cbElements.put(clusterId, cbElement);
				} else {
					cbElement.addToSum(user.getCbElements()[clusterId].getRatingsSum());
					cbElement.increment(user.getCbElements()[clusterId].getRatingsCounter());
				}
			}
			
		}
		
		int[] usersArray = new int[usersList.size()];
		int i = 0;
		for (Integer userId : usersList) {
			usersArray[i] = userId;
			i++;
		}
		
		int maxItemId = Collections.max(cbElements.keySet());
		CBElement[] cbElementsArray = new CBElement[maxItemId + 1];
		for (Entry<Integer,CBElement> entry : cbElements.entrySet()) {
			cbElementsArray[entry.getKey()] = entry.getValue();
		}
		
		UserMapOutputValueWritable valueOut = new UserMapOutputValueWritable(usersArray, cbElementsArray);
		
		context.write(key, valueOut);
		
	}

}
